package com.aotain.nyx.cc;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Base64;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.apache.log4j.Logger;

import com.aotain.common.mongo.DataColumn;
import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.nyx.common.URLStatTuple;
import com.aotain.nyx.statis.AbnStatisTuple;

/**
 * CC 攻击检测Flink 实现
 * @author Administrator
 *
 */
public class CCDetectFlink implements FlatMapFunction<Tuple2<String,URLStatTuple>, AbnStatisTuple>{
	
	/**
	 * 连接次数门限 50次/秒
	 */
	private final int ATTACK_THRESHOLD = 50*60;
	
	/**
	 * 存放记录连续状态 最后的更新时间
	 */
	private HashMap<String,Long> mapLastTime = new HashMap<String,Long>();
	
	/**
	 * 记录攻击开始时间
	 */
	private HashMap<String,Long> mapAttackStart = new HashMap<String,Long>();
	
	//-----累积包，次数
	/**
	 * 持久-上行包字节数
	 */
	private HashMap<String,Long> mapStreamoctetsMem = new HashMap<String,Long>();
	
	/**
	 * 持久-上行包数
	 */
	private HashMap<String,Long> mapStreampacketMem = new HashMap<String,Long>();

	/**
	 * 持久-攻击次数
	 */
	private HashMap<String,Long> mapCCNumMem = new HashMap<String,Long>();
	
	/**
	 * 持续攻击时记录第一次开始的Map拼接的EntryKey
	 */
	private HashMap<String,String> mapEntryKey = new HashMap<String,String>();
	
	/**
	 * 攻击峰值
	 */
	private HashMap<String,Double> mapHighFlow = new HashMap<String,Double>();
	
	/**
	 * 每分钟累积MAP  目标|源|URL
	 */
	private HashMap<String,Long> mapUrl = new HashMap<String,Long>();
	
	/**
	 * 上行包字节数
	 */
	private HashMap<String,Long> mapStreamoctets = new HashMap<String,Long>();
	
	/**
	 * 上行包数
	 */
	private HashMap<String,Long> mapStreampacket = new HashMap<String,Long>();
	
	/**
	 * 结束等待时长
	 */
	private final long END_DURATION = 180*1000l;

	/**
	 * 
	 */
	private static final long serialVersionUID = 159807604226101910L;

	@Override
	public void flatMap(Tuple2<String, URLStatTuple> arg0, Collector<AbnStatisTuple> arg1)
			throws Exception {
		// TODO Auto-generated method stub
		try {
			
				if(arg0.f0 == null) {
					return;
				}
				String[] tuplesp = arg0.f0.split("\\|",-1);
				String dip = tuplesp[0];
				String sip = tuplesp[1];
				String url = tuplesp[2];
				
				
				String ua = tuplesp[5];
				String cookie = tuplesp[6];
				String hashKey = dip + "_" + sip + "_" + url + "_" + ua + "_" + cookie;
				
				URLStatTuple info = arg0.f1;
				//System.out.println("PRINT:" + hashKey + "  NUM:" + info.getCCNum());
				mapStreamoctets.put(arg0.f0, info.getUpStreamOctets());
				mapStreampacket.put(arg0.f0, info.getUpStreamPacket());
//				String tableName = "METIS_ABNORMAL_LOG";
				//超出门限
				if(info.getHTTPNum() >= ATTACK_THRESHOLD)
				{	
					//同一个源IP 到目的IP 相同的URL 在10秒内访问次数超过100次 
					//判断当前源->目的IP是否存在记录
					if(mapLastTime.containsKey(hashKey))
					{
						//存在记录，更新数据
						mapLastTime.put(hashKey, System.currentTimeMillis());
						Logger.getLogger(CCDetectFlink.class).info("###CC V2 ATTACK UPDATE LASTTIME" + hashKey);
						//System.out.println("###CC V2 ATTACK UPDATE LASTTIME" + hashKey);
					
						AbnStatisTuple abnTuple = UpdateAttackLog(arg0.f0,info.getHTTPNum());
//						MongoSinkObject sinkObj = UpdateAttackLog(arg0.f0,info.getCCNum());
						arg1.collect(abnTuple);
						
					}
					else
					{
						//新发生的攻击行为，记录开始时间并且上报异常攻击日志，但是不上报统计数据
						mapLastTime.put(hashKey, System.currentTimeMillis());
						AbnStatisTuple abnTuple = InsertNewAttackLog(arg0.f0,info,info.getHTTPNum());
						abnTuple.setAttackNum(1l);
//						MongoSinkObject sinkObj = new MongoSinkObject();
//						sinkObj.setDmlType(DML.INSERT);
//						sinkObj.setTableName(tableName);
//						sinkObj.setImportData(row);
						arg1.collect(abnTuple);
						Logger.getLogger(CCDetectFlink.class).info("###CC V2 ATTACK START" + hashKey);
						//System.out.println("###CC V2 ATTACK START" + hashKey);
					}
				}
				else
				{
					if(mapLastTime.containsKey(hashKey))
					{
						//如果存在记录，判断更新时间与当前时间差距是否超过了等待门限
						Long lastTime = mapLastTime.get(hashKey);
						if(System.currentTimeMillis() - lastTime > END_DURATION)
						{
							//超过30秒，认为攻击结束，记录结束时间，
							Logger.getLogger(CCDetectFlink.class).info("###CC V2 ATTACK END" + hashKey);
							//System.out.println("###CC V2 ATTACK END" + hashKey);
							AbnStatisTuple abnTuple = EndAttackLog(arg0.f0);
							arg1.collect(abnTuple);
							//将数据从map中移除
							//mapWUrl.remove(base64Key);
							mapLastTime.remove(hashKey);
							mapStreamoctetsMem.remove(hashKey);
							mapStreampacketMem.remove(hashKey);
							mapCCNumMem.remove(hashKey);
							mapAttackStart.remove(hashKey);
							mapEntryKey.remove(hashKey);
							mapHighFlow.remove(hashKey);
							//Logger.getLogger(CCDetectBoltV2.class).info("###CC V2 ATTACK END" + entryKey);
						}
						else
						{
							//虽然未超过门限，但是在攻击等待范围时间内，将这一分钟的数据累加到攻击数据中。
							//UpdateAttackLog(entry);
						}
					}
				}
				
				// 对内存中存在的正在监控的攻击数据做核查判断是否定义攻击结束
				ArrayList<String> removeItem = new ArrayList<String>();
				for(Entry<String, Long> entry : mapLastTime.entrySet())
				{
					Long lastTime = entry.getValue();
					String lastTimeKey = entry.getKey();
//					Logger.getLogger(CCDetectFlink.class).info(
//									String.format("###CC V2 REMOVE hashkey %s time %d", lastTimeKey, 
//											(System.currentTimeMillis() - lastTime)/1000));
					
				
					if(System.currentTimeMillis() - lastTime > END_DURATION)
					{
						//超过等待门限秒，认为攻击结束，记录结束时间，
						String entryKey = mapEntryKey.get(lastTimeKey);
						Logger.getLogger(CCDetectFlink.class).info("###CC V2 ATTACK END" + entryKey);
						
						AbnStatisTuple abnTuple = EndAttackLog(entryKey);
						arg1.collect(abnTuple);
						//将数据从map中移除
						//mapWUrl.remove(base64Key);
						removeItem.add(lastTimeKey);
						mapStreamoctetsMem.remove(lastTimeKey);
						mapStreampacketMem.remove(lastTimeKey);
						mapCCNumMem.remove(lastTimeKey);
						mapAttackStart.remove(lastTimeKey);
						mapEntryKey.remove(lastTimeKey);
						mapHighFlow.remove(lastTimeKey);
								
					}
				}
						
				//remove 已经攻击结束的记录
				for(String base64Key : removeItem)
				{
					mapLastTime.remove(base64Key);
				}
						
				removeItem.clear();
				mapStreamoctets.clear();
				mapStreampacket.clear();
				mapUrl.clear();
				
		} catch (Exception e) {
					// TODO Auto-generated catch block
			e.printStackTrace();
			Logger.getLogger(CCDetectFlink.class).error("CCDetectFlink emitCountingData throws an exception!",e);
		}
		
	}
	
	private AbnStatisTuple InsertNewAttackLog(String entryKey,URLStatTuple tuple,Long ccNum)
	{
		String[] tuplesp = entryKey.split("\\|",-1);
		String dip = tuplesp[0];
		String sip = tuplesp[1];
		String url = tuplesp[2];
		
		String gis = new String(Base64.decodeBase64(tuplesp[3]));
		String[] gisArray = gis.split("#",-1);
		String idc = tuplesp[4];
		String ua = tuplesp[5];
		String cookie = tuplesp[6];
		
		String destAreaName = gisArray[0];
		String destGis = gisArray[1];
		String sourceAreaName = gisArray[2];
		String sourceGis = gisArray[3];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];							
		String sourceProvinceName =  gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
		String sourceAreaCityId = gisArray[7];
		String sourceAreaProvinceId = gisArray[8];
		
		String hashKey = dip + "_" + sip + "_" + url + "_" + ua + "_" + cookie;
		//String base64Key = new String(Base64.encodeBase64(hashKey.getBytes()));
		String base64Key = hashKey; 
		//url = new String(Base64.decodeBase64(url));	
		
		long starttime = System.currentTimeMillis();
		mapAttackStart.put(base64Key,starttime);//记录开始时间
		
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(starttime));
		String rowKeyAbnormal =  accesstime + "_" + sip + "_" + dip +  "_" + "3";	

		
		String desc = String.format("WEB攻击 ,攻击URL:%s", url);
		
		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
//		MongoConnection conn = MongoConnection.getInstance();
//		List<List<DataColumn>> data = new ArrayList<List<DataColumn>>();
		List<DataColumn> row = new ArrayList<DataColumn>();
		
		row.add(new DataColumn("ROWKEY",rowKeyAbnormal)); //主键字段，用于更新记录
		row.add(new DataColumn("SOURCEIP",sip));
		row.add(new DataColumn("DESTPORT",""));
		row.add(new DataColumn("ACCESSTIME",Long.parseLong(accesstime)));
		row.add(new DataColumn("ABRNORMAL", 3));
		row.add(new DataColumn("DESTIP", dip));
		row.add(new DataColumn("SOURCEAREA", sourceAreaName));
		row.add(new DataColumn("SOURCEGEO", sourceGis));
		row.add(new DataColumn("SOURCECOUNTRY", sourceAreaCountry));
		row.add(new DataColumn("DESTAREA", destAreaName));
		row.add(new DataColumn("DESTGEO", destGis));
		row.add(new DataColumn("ATTNUM", ccNum));
		row.add(new DataColumn("DESC", desc));
		row.add(new DataColumn("EVALUATE", "40"));
		row.add(new DataColumn("PROVINCE", sourceProvinceName));
		row.add(new DataColumn("ATTTYPE", "2"));
		row.add(new DataColumn("FLOWDIRECTION", tuple.getFlowDirection()));
		
		long countValue = ccNum;

		//UDII数据特有
		long nUpStreamOctets = 0;
		long nUpstreampacket = 0;
		
		if(mapStreamoctets.size() > 0 && mapStreampacket.size() > 0)
		{
			nUpStreamOctets = mapStreamoctets.get(entryKey);
			nUpstreampacket = mapStreampacket.get(entryKey);
//			hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(nUpStreamOctets));
//			hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(nUpstreampacket));
			row.add(new DataColumn("UPSTREAMOCTETS", nUpStreamOctets));
			row.add(new DataColumn("UPSTREAMPACKET", nUpstreampacket));
			
			mapStreamoctetsMem.put(base64Key, nUpStreamOctets);
			mapStreampacketMem.put(base64Key, nUpstreampacket);
		}
		mapCCNumMem.put(base64Key, countValue);
		
		mapEntryKey.put(base64Key, entryKey);
		
		importtool.InsertRowData(tableName, row);
		
		double maxValue =  nUpStreamOctets/(1024*60);
		mapHighFlow.put(rowKeyAbnormal, maxValue);
		
		AbnStatisTuple abnTuple = new AbnStatisTuple();
		abnTuple.setDestIP(dip);
		abnTuple.setSourceIP(sip);
		abnTuple.setType("CC");
		abnTuple.setAttackNum(1l);
		abnTuple.setUpStreamOctets(nUpStreamOctets);
		abnTuple.setUpStreamPacket(nUpstreampacket);
		abnTuple.setSourceCountry(sourceAreaCountry);
		abnTuple.setSourceProvince(sourceProvinceName);
		abnTuple.setAttackType("2");
//		abnTuple.setAttackMax(maxValue);
		
		Logger.getLogger(CCDetectFlink.class).info(
				String.format("###CC V2 ATTACK START ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, hashKey));
		return abnTuple;
	}
	
	private AbnStatisTuple UpdateAttackLog(String entryKey,Long ccNum)
	{
		String[] tuplesp = entryKey.split("\\|",-1);
		String dip = tuplesp[0];
		String sip = tuplesp[1];
		String url = tuplesp[2];
		
		
		String ua = tuplesp[5];
		String cookie = tuplesp[6];
		String hashKey = dip + "_" + sip + "_" + url + "_" + ua + "_" + cookie;
		String base64Key = hashKey;
	
		//UDII数据特有
		long nUpStreamOctets = 0;
		long nUpstreampacket = 0;
		if(mapStreamoctets.size() > 0 && mapStreampacket.size() > 0)
		{
			nUpStreamOctets = mapStreamoctets.get(entryKey);
			nUpstreampacket = mapStreampacket.get(entryKey);
			if(mapStreamoctetsMem.containsKey(base64Key))
				mapStreamoctetsMem.put(base64Key, mapStreamoctetsMem.get(base64Key) + nUpStreamOctets);
			if(mapStreampacketMem.containsKey(base64Key))
				mapStreampacketMem.put(base64Key, mapStreampacketMem.get(base64Key) + nUpstreampacket);
		}
		
		long countValue = ccNum;
		if(mapCCNumMem.containsKey(base64Key))
			mapCCNumMem.put(base64Key, mapCCNumMem.get(base64Key) + countValue);
		
		String desc = String.format("WEB攻击 ,攻击URL[%s]", url);
		double maxValue = (double)nUpStreamOctets/(double)60/(double)1024 ;
		
		
		if(mapHighFlow.containsKey(base64Key))
		{
			double maxValueMem = mapHighFlow.get(base64Key);
			if(maxValueMem < maxValue)
			{
				maxValueMem = maxValue;
				mapHighFlow.put(base64Key, maxValueMem);
			}
			else
			{
				maxValue = maxValueMem;
			}
		}
		else
		{
			mapHighFlow.put(base64Key, maxValue);
		}
		
		desc = String.format("%s 攻击峰值为 :%f KB/s ", desc,maxValue);
		
//		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();
		
		//开始时间
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(mapAttackStart.get(base64Key)));
				
		String rowKeyAbnormal =  accesstime + "_" + sip + "_" + dip +  "_" + "3";	
				
		if(mapStreamoctets.size() > 0 && mapStreampacket.size() > 0)
		{
			nUpStreamOctets = mapStreamoctetsMem.get(base64Key);
			nUpstreampacket = mapStreampacketMem.get(base64Key);
			row.add(new DataColumn("UPSTREAMOCTETS",nUpStreamOctets));
			row.add(new DataColumn("UPSTREAMPACKET",nUpstreampacket));
//			hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(nUpStreamOctets));
//			hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(nUpstreampacket));
			desc = String.format("%s 攻击流量[%f KB]",desc, (double)nUpStreamOctets/(double)1024);
		}
		
		
		desc = String.format("%s 攻击次数 [%d]",desc, countValue);
		Logger.getLogger(CCDetectFlink.class).info(
				String.format("###CC V2 ATTACK UPDATE ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, hashKey));
//		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "DESC", desc);
//		hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "ATTNUM", String.valueOf(countValue));
		row.add(new DataColumn("DESC",desc));
		row.add(new DataColumn("ATTNUM",countValue));
		
		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(tableName, filter, row);
		
		
		String gis = new String(Base64.decodeBase64(tuplesp[3]));
		String[] gisArray = gis.split("#",-1);
		String sourceAreaName = gisArray[2];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];							
		String sourceProvinceName =  gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
		String idc = tuplesp[4];
		
		Date current = new Date(System.currentTimeMillis());
		SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
		SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
		String dateStr_h = sdf_h.format(current);
		String dateStr_d = sdf_d.format(current);
		String dateStr_hour = sdf_hour.format(current);
		try {
			nUpStreamOctets = mapStreamoctets.get(entryKey);
			nUpstreampacket = mapStreampacket.get(entryKey);
			if(nUpStreamOctets != 0)
				countValue = nUpStreamOctets;//UDII数据用于趋势图展示
		} catch (Exception ex) {
			System.out.println("ERROR");
		}
		
		AbnStatisTuple abnTuple = new AbnStatisTuple();
		abnTuple.setDestIP(dip);
		abnTuple.setSourceIP(sip);
		abnTuple.setType("CC");
		abnTuple.setUpStreamOctets(countValue);
		abnTuple.setUpStreamPacket(nUpstreampacket);
		abnTuple.setSourceCountry(sourceAreaCountry);
		abnTuple.setSourceProvince(sourceProvinceName);
		abnTuple.setAttackType("2");
//		abnTuple.setAttackMax(maxValue);
		
		return abnTuple;	
	}

	private AbnStatisTuple EndAttackLog(String entryKey)
	{
		//UpdateAttackLog(entry);
		String[] tuplesp = entryKey.split("\\|",-1);
		String dip = tuplesp[0];
		String sip = tuplesp[1];
		String url = tuplesp[2];
		
		String gis = new String(Base64.decodeBase64(tuplesp[3]));
		String[] gisArray = gis.split("#",-1);
		String idc = tuplesp[4];
		String ua = tuplesp[5];
		String cookie = tuplesp[6];
		
		String sourceAreaName = gisArray[2];
		String sourceAreaCountry = gisArray[4];
		String sourceAreaId = gisArray[5];							
		String sourceProvinceName =  gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
		
		
		String hashKey = dip + "_" + sip + "_" + url + "_" + ua + "_" + cookie;
		String base64Key = hashKey;
		
		//开始时间
		String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(mapAttackStart.get(base64Key)));
		String endtime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(mapLastTime.get(base64Key)+60*1000l));
		
		String rowKeyAbnormal =  accesstime + "_" + sip + "_" + dip +  "_" + "3";	

		MongoImportTool importtool = MongoImportTool.getInstance();
		String tableName = "SDS_ABNORMAL_LOG2";
		List<DataColumn> row = new ArrayList<DataColumn>();
		
		String desc = String.format("WEB攻击 ,攻击URL[%s]", url);
		
		row.add(new DataColumn("ATTNUM", mapCCNumMem.get(base64Key)));
		row.add(new DataColumn("ENDTIME", endtime));
		
		desc = String.format("%s 攻击时间[%s->%s] 攻击时长[%s秒]", desc, accesstime, endtime ,
					(mapLastTime.get(base64Key)+60*1000l-mapAttackStart.get(base64Key))/1000);
		
		long nUpStreamOctets = 0;
		long nUpstreampacket = 0;
		if(mapStreamoctets.size() > 0 && mapStreampacket.size() > 0)
		{
			nUpStreamOctets = mapStreamoctetsMem.get(base64Key);
			nUpstreampacket = mapStreampacketMem.get(base64Key);
			row.add(new DataColumn("UPSTREAMOCTETS",nUpStreamOctets));
			row.add(new DataColumn("UPSTREAMPACKET",nUpstreampacket));
//			hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(nUpStreamOctets));
//			hbaseInstance.Add("METIS_ABNORMAL_LOG", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(nUpstreampacket));
			desc = String.format("%s 攻击流量[%f KB]",desc, (double)nUpStreamOctets/(double)1024);
		}
		
		long countValue = mapCCNumMem.get(base64Key);
		desc = String.format("%s 攻击次数 [%d]",desc, countValue);
		
		if(mapHighFlow.containsKey(base64Key))
		{
			double maxValueMem = mapHighFlow.get(base64Key);
			desc = String.format("%s 攻击峰值 [%f KB/s]",desc, maxValueMem);
		}


		row.add(new DataColumn("DESC",desc));
		row.add(new DataColumn("ATTNUM",countValue));

		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("ROWKEY",rowKeyAbnormal));
		importtool.UpdateRowData(tableName, filter, row);
		
		nUpStreamOctets = mapStreamoctets.get(entryKey);
		nUpstreampacket = mapStreampacket.get(entryKey);
		
		AbnStatisTuple abnTuple = new AbnStatisTuple();
		abnTuple.setDestIP(dip);
		abnTuple.setSourceIP(sip);
		abnTuple.setType("CC");
		abnTuple.setUpStreamOctets(nUpStreamOctets);
		abnTuple.setUpStreamOctets(nUpstreampacket);
		abnTuple.setAttTime(accesstime);
		abnTuple.setAttackType("2");
		
		
		Logger.getLogger(CCDetectFlink.class).info(
				String.format("###CC V2 ATTACK END ROWKEY:%s DESC:%s  HASHKEY:%s",rowKeyAbnormal, desc, hashKey));

		return abnTuple;
	}
}
